<?php
// 消费回调队列，获取处理结果
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$redis->select(1);

$callbackStream = 'task:callback:queue';
$groupName = 'callback_group';
$consumerName = 'callback_worker';

// 创建回调队列消费者组（只需一次）
try {
    $redis->xGroup('CREATE', $callbackStream, $groupName, '0', true);
} catch (RedisException $e) {}

while (true) {
    $tasks = $redis->xReadGroup($groupName, $consumerName, [$callbackStream => '>'], 1, 1000);
    if (!empty($tasks[$callbackStream])) {
        foreach ($tasks[$callbackStream] as $msgId => $fields) {
            echo "收到回调结果: 任务ID={$fields['task_id']} 结果={$fields['result']}\n";
            $redis->xAck($callbackStream, $groupName, [$msgId]);


        }
        // 删除已处理的任务
        $deletedCount = $redis->xDel($callbackStream, [$msgId]);
        echo "已删除任务: $msgId (删除数量: $deletedCount)\n";
    }
    usleep(500000); // 0.5秒
} 